package cn.whuc.sparkstreaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object Demo_ncToStreaming {
  def main(args: Array[String]): Unit = {
    // 1 创建配置文件对象
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("streaming")
    // 2 创建sparkstreaming程序入口,这个新的入口需要一个时间间隔 时间窗口
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))
    // 3 流式数据处理的过程
    val value: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)

    // 4 通过原语 wordcount
    val words: DStream[String] = value.flatMap(_.split(" "))

    val wordToOne: DStream[(String, Int)] = words.map((_, 1))

    val wordCount: DStream[(String, Int)] = wordToOne.reduceByKey(_ + _)

    wordCount.print()

    // 5 什么时候，当前程序开始 接受 nc的数据呢？
    ssc.start()
    ssc.awaitTermination()
    // 有没有关闭程序？ 不好意思 没有 流式数据 7*24
  }
}
